package com.meiya.flink.util;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

import org.apache.flink.core.execution.JobClient;
import org.apache.flink.table.api.TableResult;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 异步SQL工具类，用于处理异步SQL执行和打印操作
 */
public class AsyncSqlUtil {
    private static final Logger logger = LoggerFactory.getLogger(AsyncSqlUtil.class);
    
    // 线程计数器，用于生成唯一的线程名
    private static final AtomicInteger threadCounter = new AtomicInteger(0);
    
    // 活动线程集合，用于跟踪和管理线程
    private static final Set<Thread> activeThreads = Collections.synchronizedSet(new HashSet<>());
    
    // 添加JVM关闭钩子，确保所有线程在JVM关闭时被中断
    static {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            logger.info("JVM关闭，中断所有活动线程...");
            interruptAllThreads();
            logger.info("所有活动线程已中断");
        }));
    }
    
    /**
     * 使用Thread异步执行SQL查询，不等待结果
     * 
     * @param sql SQL语句
     * @param executor SQL执行器接口
     * @param operationName 操作名称（用于日志）
     */
    public static void executeAsyncWithThread(
            String sql, 
            SqlExecutor executor,
            String operationName) {
        
        logger.info("提交[ {} ]操作: [ {} ]", operationName, sql);
        
        try {

            getTableResult(sql, executor, operationName);

            // 创建线程来处理结果（如果需要）
            Thread resultThread = new Thread(() -> {
                try {
                    // 这里可以添加结果处理逻辑，但不调用result.print()，避免阻塞
                    logger.info("[ {} ]作业结果处理线程启动", operationName);
                    // 可以在这里添加其他结果处理逻辑
                } catch (Exception e) {
                    logger.error("处理[ {} ]作业结果失败: [ {} ]", operationName, e.getMessage(), e);
                } finally {
                    // 从活动线程集合中移除当前线程
                    activeThreads.remove(Thread.currentThread());
                    logger.debug("结果处理线程已完成，当前活动线程数: [ {} ]", activeThreads.size());
                }
            });
            
            // 设置线程名称
            resultThread.setName(operationName + "-result-thread-" + threadCounter.incrementAndGet());
            
            // 设置为守护线程，不阻止JVM退出
            resultThread.setDaemon(true);
            
            // 添加到活动线程集合
            activeThreads.add(resultThread);
            
            // 启动线程
            logger.info("启动[ {} ]结果处理线程: [ {} ]", operationName, resultThread.getName());
            resultThread.start();
            
            logger.info("[ {} ]结果处理线程已启动，当前活动线程数: [ {} ]", operationName, activeThreads.size());
            
        } catch (Exception e) {
            logger.error("执行[ {} ]SQL失败: [ {} ]", operationName, e.getMessage(), e);
        }
    }

    /**
     * 中断所有活动线程
     */
    public static void interruptAllThreads() {
        synchronized (activeThreads) {
            logger.info("中断所有活动线程，当前活动线程数: [ {} ]", activeThreads.size());
            for (Thread thread : activeThreads) {
                try {
                    if (thread.isAlive()) {
                        logger.info("中断线程: [ {} ]", thread.getName());
                        thread.interrupt();
                    }
                } catch (Exception e) {
                    logger.error("中断线程[ {} ]时发生错误: [ {} ]", thread.getName(), e.getMessage(), e);
                }
            }
            activeThreads.clear();
        }
    }

    /**
     * 检查表是否存在
     * 
     * @param tableName 表名
     * @param executor SQL执行器接口
     * @return 表是否存在
     */
    public static boolean checkTableExists(String tableName, SqlExecutor executor) {
        try {
            // 使用DESCRIBE语句检查表是否存在，这通常不会触发实际的查询执行
            executor.executeSql("DESCRIBE " + tableName);
            logger.info("表[ {} ]存在", tableName);
            return true;
        } catch (Exception e) {
            logger.error("表[ {} ]不存在或结构有误: [ {} ]", tableName, e.getMessage());
            return false;
        }
    }

    @NotNull
    private static TableResult getTableResult(String sql, SqlExecutor executor, String operationName) {

        final TableResult result = executor.executeSql(sql);
        logger.info("[ {} ]SQL执行成功，获取JobClient", operationName);

        // 获取JobClient但不等待作业完成
        if (result.getJobClient().isPresent()) {
            JobClient jobClient = result.getJobClient().get();
            logger.info("[ {} ]作业已提交，作业ID: [ {} ]", operationName, jobClient.getJobID());
        } else {
            logger.info("[ {} ]作业已提交，但无法获取作业ID", operationName);
        }
        return result;
    }

    /**
     * SQL执行器接口
     */
    @FunctionalInterface
    public interface SqlExecutor {
        /**
         * 执行SQL语句
         * 
         * @param sql SQL语句
         * @return 表结果
         */
        TableResult executeSql(String sql);
    }

    /**
     * 在禁用JMX的环境中执行操作
     * 
     * @param <T> 操作结果类型
     * @param operation 要执行的操作
     * @return 操作结果
     */
    public static <T> T executeWithJmxDisabled(Supplier<T> operation) {
        // 保存原始JMX设置
        String originalJmxDisabled = System.getProperty("org.apache.kafka.disableJmx");
        
        try {
            // 禁用JMX
            System.setProperty("org.apache.kafka.disableJmx", "true");
            
            // 执行操作
            return operation.get();
        } finally {
            // 恢复原始JMX设置
            if (originalJmxDisabled != null) {
                System.setProperty("org.apache.kafka.disableJmx", originalJmxDisabled);
            } else {
                System.clearProperty("org.apache.kafka.disableJmx");
            }
        }
    }
    
    /**
     * 在禁用JMX的环境中执行无返回值的操作
     * 
     * @param operation 要执行的操作
     */
    public static void runWithJmxDisabled(Runnable operation) {
        // 保存原始JMX设置
        String originalJmxDisabled = System.getProperty("org.apache.kafka.disableJmx");
        
        try {
            // 禁用JMX
            System.setProperty("org.apache.kafka.disableJmx", "true");
            
            // 执行操作
            operation.run();
        } finally {
            // 恢复原始JMX设置
            if (originalJmxDisabled != null) {
                System.setProperty("org.apache.kafka.disableJmx", originalJmxDisabled);
            } else {
                System.clearProperty("org.apache.kafka.disableJmx");
            }
        }
    }

    /**
     * 将TableResult转换为字符串
     * 
     * @param result 表结果
     * @return 表结果的字符串表示
     */
    public static String tableResultToString(TableResult result) {
        if (result == null) {
            return "null";
        }
        
        StringBuilder sb = new StringBuilder();
        
        try {
            // 获取列名
            String[] columnNames = result.getTableSchema().getFieldNames();
            sb.append("Columns: ").append(String.join(", ", columnNames)).append("\n");
            
            // 尝试获取前10行数据
            sb.append("Data (max 10 rows):\n");
            
            // 添加超时机制
            CompletableFuture<Void> fetchFuture = CompletableFuture.runAsync(() -> {
                try {
                    int rowCount = 0;
                    result.collect().forEachRemaining(row -> {
                        if (rowCount < 10) {
                            sb.append(row.toString()).append("\n");
                        }
                    });
                } catch (Exception e) {
                    sb.append("Error fetching data: ").append(e.getMessage());
                }
            });
            
            try {
                // 设置5秒超时
                fetchFuture.get(5, TimeUnit.SECONDS);
            } catch (TimeoutException e) {
                sb.append("Timeout while fetching data");
                fetchFuture.cancel(true);
            } catch (Exception e) {
                sb.append("Error: ").append(e.getMessage());
            }
            
        } catch (Exception e) {
            sb.append("Error converting TableResult to string: ").append(e.getMessage());
        }
        
        return sb.toString();
    }
} 